package com.bds.btcdc.adapter.okcoin;

import com.alibaba.fastjson.JSONObject;
import com.bds.btcdc.model.CoinPairExchanges;
import com.bds.btcdc.model.MarketData;
import com.bds.btcdc.api.model.Result;
import com.bds.btcdc.util.Const;
import com.bds.btcdc.util.Utils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompSession;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * OkCoin的WebSocket客戶端，接受OkCoin的行情更新推送
 */
public class OkCoinReceiver extends WebSocketClient {
	private Logger logger = LoggerFactory.getLogger(getClass());
	private List<CoinPairExchanges> okcoin_pairs;
	private static final String EXCHANGE = "OKCOIN";
	private StompSession session;

	public OkCoinReceiver(StompSession session, URI serverURI, List<CoinPairExchanges> okcoin_pairs) {
		super(serverURI);
		this.session = session;
		this.okcoin_pairs = okcoin_pairs;
	}

	public OkCoinReceiver(URI serverUri, Draft draft) {
		super(serverUri, draft);
	}

	public OkCoinReceiver(URI serverURI) {
		super(serverURI);
	}

	@Override
	public void onOpen(ServerHandshake handshakedata) {
		// 批量订阅
		// "[{'event':'addChannel','channel':'ok_sub_spot_btc_usd_ticker'},{'event':'addChannel','channel':'ok_sub_spot_usd_btc_depth'}]";
		StringBuilder sb = new StringBuilder();
		for (CoinPairExchanges pair_exchanges : okcoin_pairs) {
			String dataMsg = "{'event':'addChannel','channel':'ok_sub_spot_" + pair_exchanges.getPair_code()
					+ "_ticker'}";
			sb.append(dataMsg + ",");
		}
		String msgss = sb.toString().substring(0, sb.length() - 1);
		msgss = "[" + msgss + "]";
		send(msgss);
		logger.info("######### OkCoin 连接成功 #############");
	}

	@Override
	public void onClose(int code, String reason, boolean remote) {
		if (code == -1) {
			logger.info("######### OkCoin连接断开，重连中 ######## 断开原因：" + reason);
			new Thread() {
				@Override
				public void run() {
					try {
						Thread.sleep(Const.RE_CONN_TIME);
						new OkCoinReceiver(session, new URI(Const.OKCOIN.WS_URL), okcoin_pairs).connect();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}.start();
		}
	}

	@Override
	public synchronized void onMessage(String message) {
//		 logger.info("行情更新: " + message);

		Result ret = new Result();
		List<MarketData> ms = new ArrayList<MarketData>();
		message = Utils.regex("\\[(.*?)\\]", message);
		JSONObject msg = JSONObject.parseObject(message);
		String dataStr = msg.getString("data");
		MarketData data = JSONObject.parseObject(dataStr, MarketData.class);
		if (data.getBuy() != null) {
			String channel = msg.getString("channel");
			String pair = Utils.regex("ok_sub_spot_(.*?)_ticker", channel);
			data.setExchange(EXCHANGE);
			data.setPair(pair);
			ms.add(data);
			ret.setCode(Const.SUC);
			ret.setData(ms);
			if (session != null)
				// 推到btc-web主题

				try {
					session.send(Const.OKCOIN.MARKET_TOPIC, JSONObject.toJSONString(ret).getBytes());
				} catch (Exception e) {
					logger.error("", e);
				}
		}
	}

	@Override
	public void onMessage(ByteBuffer message) {
		logger.info("received ByteBuffer");
	}

	@Override
	public void onError(Exception ex) {
		logger.info("an error occurred:" + ex);
	}

	// 测试Receiver
	public static void main(String[] args) throws Exception {
		new OkCoinReceiver(new URI("wss://real.okcoin.cn:10440/websocket")).connect();
	}
}